package com.xxk.rocket.star.core.client.impl;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.MQConsumer;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.xxk.rocket.star.core.client.AbsProcessClientConsume;
import com.xxk.rocket.star.core.dto.StargateConsumeBaseDTO;

public class ConsumeBaseProcessClientConsume extends
        AbsProcessClientConsume<StargateConsumeBaseDTO> {

  @Override
  protected void processConsume(StargateConsumeBaseDTO stargateConsumeBaseDTO,
                                MQConsumer mqConsumer) throws Throwable {
    DefaultMQPushConsumer defaultMQPushConsumer = (DefaultMQPushConsumer) mqConsumer;
    defaultMQPushConsumer.setPullBatchSize(stargateConsumeBaseDTO.getPullBatchSize());
    defaultMQPushConsumer.setConsumeFromWhere(stargateConsumeBaseDTO.getConsumeFromWhere());
    if (stargateConsumeBaseDTO.isBroadcasting()) {
      defaultMQPushConsumer.setMessageModel(MessageModel.BROADCASTING);
    } else {
      defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
    }
  }
}
